package com.mchat.user.service.impl;


import com.huaban.analysis.jieba.JiebaSegmenter;
import com.huaban.analysis.jieba.SegToken;
import com.mchat.config.JiebaSegmenterSingleton;
import com.mchat.config.JwtConfig;
import com.mchat.constant.RedisConstant;
import com.mchat.constant.UserState;
import com.mchat.exceptions.CommonException;
import com.mchat.user.dao.UserDAO;
import com.mchat.user.domain.constant.AccountStatus;
import com.mchat.user.domain.constant.Gender;
import com.mchat.user.domain.dto.*;
import com.mchat.user.domain.mapper.UserMapper;
import com.mchat.user.domain.po.User;
import com.mchat.user.domain.vo.CurrentUserVo;
import com.mchat.user.domain.vo.LoginUserVO;
import com.mchat.user.domain.vo.UserVO;
import com.mchat.user.service.IUserService;
import com.mchat.util.JsonUtil;
import com.mchat.util.MailUtil;
import com.mchat.util.SnowflakeIdGenerator;
import io.jsonwebtoken.Claims;
import io.jsonwebtoken.Jwts;
import io.jsonwebtoken.SignatureAlgorithm;
import io.jsonwebtoken.security.Keys;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.reactivestreams.Publisher;
import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.types.Expiration;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

@Service
@RequiredArgsConstructor
@Slf4j
public class UserServiceImpl implements IUserService {
    private final ReactiveStringRedisTemplate reactiveStringRedisTemplate;
    private final UserDAO userDAO;
    private final SnowflakeIdGenerator snowflakeIdGenerator;
    private final JsonUtil jsonUtil;
    private final JwtConfig jwtConfig;

    public Mono<UserVO> getUser(long id) {
        return getUserPO(id).map(UserMapper.INSTANCE::PO2VO);
    }


    private Mono<User> getUserPO(long id) {
        return reactiveStringRedisTemplate.opsForValue()
                .getAndExpire(String.valueOf(id), Duration.ofHours(1))
                .singleOptional()
                .flatMap(value -> {
                    //如果redis返回的数据为空
                    if (value.isEmpty()) {
                        return userDAO.getById(id)
                                .singleOptional()
                                .flatMap(optional -> {
                                    //user不为空
                                    if (optional.isPresent()) {
                                        User user = optional.get();
                                        return reactiveStringRedisTemplate.opsForValue()
                                                .set(String.valueOf(id), jsonUtil.toStr(user), Duration.ofHours(1))
                                                .thenReturn(user);
                                    } else {
                                        return reactiveStringRedisTemplate.opsForValue()
                                                .set(String.valueOf(id), "", Duration.ofMinutes(10))
                                                .flatMap(b -> Mono.empty());
                                    }
                                });
                    }
                    //如果redis返回的不为空但是一个不存在的值
                    if (value.get().equals("")) {
                        // 如果Redis中有数据，直接返回
                        return Mono.empty();
                    }
                    //返回redis的值
                    return Mono.just(jsonUtil.toObj(value.get(), User.class));
                });
    }
//    public Flux<UserVO> getUserBatch(List<Long> userIdCollection) {
//        if (userIdCollection == null || userIdCollection.isEmpty()) {
//            return Flux.empty();
//        }
//
//        return Flux.fromIterable(splitList(userIdCollection, 200))
//                .limitRate(1)
//                .flatMap(batchUserIds -> reactiveStringRedisTemplate
//                        .execute(
//                                connection -> Flux.fromIterable(batchUserIds)
//                                        .flatMap(id -> connection.stringCommands().getEx(ByteBuffer.wrap(String.valueOf(id).getBytes(StandardCharsets.UTF_8)), Expiration.from(Duration.ofHours(1))))
//                                        .collectList().map(v -> Tuples.of(batchUserIds, v))
//                        )
//                )
//                .flatMap(batchTuple -> {
//                    List<Long> batchUserIds = batchTuple.getT1();
//                    List<ByteBuffer> redisRawValues = batchTuple.getT2();
//
//                    //将redis的bytebuffer转化为user对象
//                    List<User> cachedUsers = new ArrayList<>(redisRawValues.size());
//                    for (ByteBuffer redisRawValue : redisRawValues) {
//                        if (redisRawValue != null) {
//                            cachedUsers.add(jsonUtil.toObj(StandardCharsets.UTF_8.decode(redisRawValue).toString(), User.class));
//                        }
//                    }
//                    //获取缓存的对象的id
//                    List<Long> cachedUserIds = new ArrayList<>(cachedUsers.size());
//                    for (User cachedUser : cachedUsers) {
//                        cachedUserIds.add(cachedUser.getId());
//                    }
//                    //获取差集
//                    List<Long> missingUserIds = batchUserIds.stream()
//                            .filter(userId -> !cachedUserIds.contains(userId))
//                            .toList();
//
//                    if (missingUserIds.isEmpty()) {
//                        return Flux.fromIterable(cachedUsers);
//                    }
//                    return userDAO.getByIds(missingUserIds)
//                            //将数据库的内容写入cache
//                            .flatMapMany(dbUsers -> reactiveStringRedisTemplate.execute(connection -> Flux.fromIterable(dbUsers).flatMap(user ->
//                                    connection.stringCommands().setEX(
//                                            ByteBuffer.wrap(String.valueOf(user.getId()).getBytes(StandardCharsets.UTF_8)),
//                                            ByteBuffer.wrap(jsonUtil.toStr(user).getBytes(StandardCharsets.UTF_8)),
//                                            Expiration.from(Duration.ofHours(1))
//                                    ))
//                            ).thenMany(Flux.concat(Flux.fromIterable(cachedUsers), Flux.fromIterable(dbUsers))));
//                }).map(UserMapper.INSTANCE::PO2VO);
//    }
    //两种不同的写法
    public Flux<UserVO> getUsers(List<Long> ids) {
        ids=ids.stream().filter(Objects::nonNull).collect(Collectors.toList());
        if (ids.size() == 0) return Flux.empty();
        return getUsers(Flux.fromIterable(ids)).map(UserMapper.INSTANCE::PO2VO);
    }
    private Flux<User> getUsers(Flux<Long> ids){
        Sinks.Many<Long> fluxSink = Sinks.unsafe().many().unicast().onBackpressureBuffer();

        Flux<User> cachedData = ids.<User>flatMap(id -> reactiveStringRedisTemplate.execute(connection ->
                                connection.stringCommands()
                                        .getEx(ByteBuffer.wrap(id.toString().getBytes()), Expiration.from(Duration.ofHours(1)))
                                        .singleOptional()
                        )
                        .handle((item, sink) -> {
                            if (item.isEmpty()) {
                                fluxSink.tryEmitNext(id);
                            } else {
                                ByteBuffer byteBuffer = item.get();
                                String s = StandardCharsets.UTF_8.decode(byteBuffer).toString();
                                sink.next(jsonUtil.toObj(s, User.class));
                            }
                        })
        ).doFinally((signalType) -> {
            fluxSink.tryEmitComplete();
        });
        Flux<User> dbData = userDAO.getBatchUser(fluxSink.asFlux())
                .concatMap(user ->
                        reactiveStringRedisTemplate.execute(connection ->
                                        connection.stringCommands().setEX(
                                                ByteBuffer.wrap(String.valueOf(user.getId()).getBytes(StandardCharsets.UTF_8)),
                                                ByteBuffer.wrap(jsonUtil.toStr(user).getBytes(StandardCharsets.UTF_8)),
                                                Expiration.from(Duration.ofHours(1))
                                        )
                                )
                                .then(Mono.just(user))
                );
        return Flux.concat(cachedData,dbData);
    }

    public Mono<LoginUserVO> register(RegisterDTO registerDTO) {
        String email = registerDTO.getEmail();
        return reactiveStringRedisTemplate.opsForValue().get(RedisConstant.VERIFY_CODE_FOR_REGISTER + ":" + email)
                .singleOptional()
                .flatMap(optional -> {
                    if (optional.isEmpty() || !optional.get().equals(registerDTO.getCode()))
                        return Mono.error(new CommonException("验证码错误"));
                    return userDAO.SelectExist(email);
                })
                .flatMap(v -> {
                    if (v) return Mono.error(new CommonException("邮箱已被注册"));
                    User user = new User();
                    long id = snowflakeIdGenerator.nextId();
                    user.setId(id);
                    user.setName("用户_" + id % 10000);
                    user.setEmail(registerDTO.getEmail());
                    // TODO: 2025年7月22日 添加密码加密,添加默认头像
                    user.setProfileImg("1.png");
                    user.setPassword(registerDTO.getPassword());
                    user.setGender(Gender.UNKNOWN);
                    user.setCreateTime(LocalDateTime.now());
                    user.setStatus(AccountStatus.NORMAL);
                    return userDAO.save(user).map(count -> {
                        if (count == 0) return new CommonException("邮箱已被注册");
                        return count;
                    }).thenReturn(user);
                })
                .flatMap(user -> {
                    LoginUserVO loginUserVO = UserMapper.INSTANCE.PO2LoginUserVO(user);
                    //生成token
                    String token = genToken(user.getId());
                    loginUserVO.setToken(token);
                    return Mono.just(loginUserVO);
                });
    }
    public Mono<LoginUserVO> login(LoginDTO loginDTO) {
        String pwd = loginDTO.getPassword();
        String email = loginDTO.getEmail();
        return userDAO.selectByEmail(email)
                .singleOptional()
                .flatMap(v -> {
                    if (v.isEmpty()) return Mono.error(new CommonException("用户不存在"));
                    User user = v.get();
                    // TODO: 2025年7月23日 添加密码加密
                    if (!user.getPassword().equals(pwd)) return Mono.error(new CommonException("密码错误", 401));
                    LoginUserVO loginUserVO = UserMapper.INSTANCE.PO2LoginUserVO(user);
                    String token = genToken(user.getId());
                    loginUserVO.setToken(token);
                    return Mono.just(loginUserVO);
                });
    }
    public Mono<Long> updateInfo(long id, UserInfoDTO dto) {
        //更新数据然后删除缓存
        return userDAO.updateInfo(id, dto)
                .flatMap(v -> reactiveStringRedisTemplate.opsForValue().delete(String.valueOf(id)).thenReturn(v))
                .map(count -> {
            if (count!=1) throw new CommonException("执行失败");
            return count;
        });
    }
    public Mono<CurrentUserVo> getCurrentUser(Long uid) {
        return userDAO.getById(uid).map(UserMapper.INSTANCE::PO2CurrentUserVO);
    }
    public Mono<Long> resetPwd(ResetPwdDTO dto) {
        String email = dto.getEmail();
        return reactiveStringRedisTemplate.opsForValue().get(RedisConstant.VERIFY_CODE_FOR_RESET_PWD + ":" + email)
                .singleOptional()
                .flatMap(optional -> {
                    if (optional.isEmpty() || !optional.get().equals(dto.getCode()))
                        return Mono.error(new CommonException("验证码错误"));
                    return userDAO.updatePwd(email, dto.getPassword());
                })
                .map(v->{
                    if (v==0) throw new CommonException("用户还未注册");
                    else if (v == 1) return v;
                    else throw new CommonException("重置失败",500);
                });
    }
    public Mono<Void> updateAvatar(long id, String avatar) {
        //修改数据库，删除缓存
        return userDAO.updateAvatar(id, avatar)
                .flatMap(
                        v -> reactiveStringRedisTemplate.opsForValue().delete(String.valueOf(id)).then()
                );
    }
    public Mono<String> updateToken(Claims claims){
        long id = Long.parseLong(claims.getSubject());
        return validUser(id,claims).map(v->{
            if (v)return genToken(id);
            throw new CommonException("token过期",401);
        });
    }
    public Mono<Boolean> validUser(long id,Claims claims){
        return userDAO.getById(id).flatMap(user -> {
            LocalDateTime lastUpdate = user.getLastUpdate();
            int status = user.getStatus();
            LocalDateTime tokenIssuedTime = claims.getIssuedAt().toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime();
            return Mono.just(tokenIssuedTime.isAfter(lastUpdate)&&status== UserState.NORMAL);
        });
    }
    private String genToken(long id) {
        //id用于标识唯一的jwt
        //sub用于标识用户id
        return Jwts.builder()
                .setId(String.valueOf(snowflakeIdGenerator.nextId()))
                .setSubject(String.valueOf(id))
                .setExpiration(new Date(System.currentTimeMillis() + jwtConfig.getExpiration()))
                .setIssuedAt(new Date())
                .signWith(Keys.hmacShaKeyFor(jwtConfig.getSecret().getBytes()), SignatureAlgorithm.HS256)
                .compact();
    }
    public Mono<Long> updatePrivacy(UserPrivacyDTO dto ,long id) {
        return userDAO.updatePrivacy(id,dto)
                .flatMap(v -> reactiveStringRedisTemplate.opsForValue().delete(String.valueOf(id)).thenReturn(v))
                .map(count -> {
                    if (count!=1) throw new CommonException("执行失败");
                    return count;
                });
    }
    public Mono<Long> updateName(long id, String name) {
        if (name==null||name.isBlank()) throw new CommonException("名称不能为空",400);
        JiebaSegmenter jiebaSegmenter = JiebaSegmenterSingleton.getInstance();
        //将名字拆分成关键字进行存储
        Mono<Long> mono;
        if(StringUtils.isAsciiPrintable(name)){
            //如果名称是ascii则使pg自带的用英文分词法
            mono=userDAO.updateName(id,name,null);
        }
        else{
            //如果名称包含中文或其他字符，则保留中文、英文、数字之后用jieba分词
            String result = name.replaceAll("[^a-zA-Z0-9\\u4e00-\\u9fa5]", "");
            mono=userDAO.updateName(id,name,jiebaSegmenter.process(result, JiebaSegmenter.SegMode.SEARCH));
        }
        return mono.flatMap(v->{
            if(v!=1)return Mono.error(new CommonException("执行失败",500));
            return reactiveStringRedisTemplate.opsForValue().delete(String.valueOf(id)).thenReturn(v);
        }
        );
    }
    public Flux<UserVO> searchUser(String keyWord){

        if (keyWord==null||keyWord.isBlank())throw new CommonException("无效的关键字",400);
        JiebaSegmenter jiebaSegmenter = JiebaSegmenterSingleton.getInstance();
        List<SegToken> tokens = jiebaSegmenter.process(keyWord, JiebaSegmenter.SegMode.SEARCH);
        StringBuilder stringBuilder = new StringBuilder(keyWord.length() * 2);
        for (int i = 0; i < tokens.size(); i++) {
            String word = tokens.get(i).word;
            if (word!=null&&!word.isBlank()&&!word.equals(" ")){
                stringBuilder.append(word);
                if(i<tokens.size()-1){
                    stringBuilder.append('&');
                }
            }
        }

        Flux<Long> ids = userDAO.searchName(stringBuilder.toString());
        Flux<User> concat;
        //如果传入的是纯数字,可能是一个id
        if(StringUtils.isNumeric(keyWord))
        {
            concat = Flux.concat( userDAO.getById(Long.parseLong(keyWord)),getUsers(ids));
        }//如果传入的是邮箱
        else if (MailUtil.isValid(keyWord)) {
            Mono<User> userMono = userDAO.selectByEmail(keyWord).filter(User::isSearchableByEmail);
            concat = Flux.concat(userMono, getUsers(ids));
        }else{
            concat=getUsers(ids);
        }
        return concat.map(UserMapper.INSTANCE::PO2VO);
    }
}
